package cn.chengpengper.rpc.consumer.common.handler;

import cn.chengpengper.rpc.constants.RpcConstants;
import cn.chengpengper.rpc.consumer.common.cache.ConsumerChannelCache;
import cn.chengpengper.rpc.consumer.common.context.RpcContext;
import cn.chengpengper.rpc.protocol.enums.RpcStatus;
import cn.chengpengper.rpc.protocol.enums.RpcType;
import cn.chengpengper.rpc.protocol.header.RpcHeaderFactory;
import cn.chengpengper.rpc.proxy.api.future.RpcFuture;
import cn.chengpengper.rpc.protocol.RpcProtocol;
import cn.chengpengper.rpc.protocol.header.RpcHeader;
import cn.chengpengper.rpc.protocol.request.RpcRequest;
import cn.chengpengper.rpc.protocol.response.RpcResponse;
import cn.hutool.json.JSONUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author chengpeng.hu
 */
@Slf4j
@Getter
public class RpcConsumerHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcResponse>> {

    private volatile Channel channel;

    private SocketAddress remotePeer;

    private Map<Long, RpcFuture> pendingRPC = new ConcurrentHashMap<>();

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer = this.channel.remoteAddress();
        ConsumerChannelCache.add(channel);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol<RpcResponse> protocol) {
        if (protocol == null) {
            return;
        }
        handlerMessage(protocol);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        super.channelUnregistered(ctx);
        ConsumerChannelCache.remove(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        ConsumerChannelCache.remove(ctx.channel());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            //发送一次心跳数据
            RpcHeader header = RpcHeaderFactory.getRequestHeader(RpcConstants.SERIALIZATION_PROTOSTUFF, RpcType.HEARTBEAT_FROM_CONSUMER);
            RpcProtocol<RpcRequest> requestRpcProtocol = new RpcProtocol<>();
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setParameters(new Object[]{RpcConstants.HEARTBEAT_PING});
            requestRpcProtocol.setHeader(header);
            requestRpcProtocol.setBody(rpcRequest);
            ctx.writeAndFlush(requestRpcProtocol);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    private void handlerMessage(RpcProtocol<RpcResponse> protocol) {
        RpcHeader header = protocol.getHeader();
        // 心跳消息
        if (header.getMsgType() == RpcType.HEARTBEAT_TO_CONSUMER.getType()) {
            this.handlerHeartbeatMessage(protocol);
        } else if (header.getMsgType() == RpcType.HEARTBEAT_FROM_PROVIDER.getType()) {
            this.handlerHeartbeatMessageFromProvider(protocol, channel);
        } else if (header.getMsgType() == RpcType.RESPONSE.getType()) {
            //响应消息
            this.handlerResponseMessage(protocol, header);
        }
    }

    private void handlerHeartbeatMessageFromProvider(RpcProtocol<RpcResponse> protocol, Channel channel) {
        log.info("【HeartBeat Provider Request Consumer】receive service provider heartbeat message:{}", protocol.getBody().getResult());
        RpcHeader header = protocol.getHeader();
        header.setMsgType(RpcType.HEARTBEAT_TO_PROVIDER.getType().byteValue());
        RpcProtocol<RpcRequest> requestRpcProtocol = new RpcProtocol<>();
        RpcRequest request = new RpcRequest();
        request.setParameters(new Object[]{RpcConstants.HEARTBEAT_PONG});
        header.setStatus(RpcStatus.SUCCESS.getCode().byteValue());
        requestRpcProtocol.setHeader(header);
        requestRpcProtocol.setBody(request);
        channel.writeAndFlush(requestRpcProtocol);
    }

    private void handlerResponseMessage(RpcProtocol<RpcResponse> protocol, RpcHeader header) {
        log.info("服务消费者接收到的响应数据====>{}", JSONUtil.toJsonPrettyStr(protocol));
        long requestId = header.getRequestId();
        RpcFuture rpcFuture = pendingRPC.remove(requestId);
        if (rpcFuture != null) {
            rpcFuture.done(protocol);
        }
    }

    private void handlerHeartbeatMessage(RpcProtocol<RpcResponse> protocol) {
        //此处简单打印即可,实际场景可不做处理
        log.info("HeartBeat【Provider Response Consumer】receive service provider heartbeat message:{}", protocol.getBody().getResult());
        // 收到服务提供者pong后，对应channel 等待数减1
        int count = ConsumerChannelCache.decreaseTimes(channel);
        log.info("服务消费者收到服务提供者【{}】心跳响应，当前心跳响应等待【{}】次", channel.remoteAddress(), count);
    }

    public RpcFuture sendRequest(RpcProtocol<RpcRequest> protocol, boolean async, boolean oneway) {
        log.info("服务消费者发送的数据====>{}", JSONUtil.toJsonStr(protocol));
        return oneway ? this.sendRequestOneway(protocol) : async ? this.sendRequestAsync(protocol) : sendRequestSync(protocol);
    }

    public RpcFuture sendRequestSync(RpcProtocol<RpcRequest> protocol) {
        RpcFuture rpcFuture = this.getRpcFuture(protocol);
        channel.writeAndFlush(protocol);
        return rpcFuture;
    }

    public RpcFuture sendRequestAsync(RpcProtocol<RpcRequest> protocol) {
        RpcFuture rpcFuture = this.getRpcFuture(protocol);
        //如果是异步调用，则将RPCFuture放入RpcContext
        RpcContext.getContext().setRpcFuture(rpcFuture);
        channel.writeAndFlush(protocol);
        return null;
    }

    public RpcFuture sendRequestOneway(RpcProtocol<RpcRequest> protocol) {
        channel.writeAndFlush(protocol);
        return null;
    }

    public RpcFuture getRpcFuture(RpcProtocol<RpcRequest> protocol) {
        RpcFuture rpcFuture = new RpcFuture(protocol);
        RpcHeader header = protocol.getHeader();
        long requestId = header.getRequestId();
        pendingRPC.put(requestId, rpcFuture);
        return rpcFuture;
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        ConsumerChannelCache.remove(channel);
    }

}
